package drds.server.net;

import java.io.IOException;
import java.nio.channels.CancelledKeyException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 网络事件反应器
 * 
 * @author 曹正辉<330937205@qq.com>
 * 
 */
public final class Reactor {
	private static final Logger LOGGER = LoggerFactory.getLogger(Reactor.class);
	private final String name;
	private final ReadWriteTask readWriteTask;

	public Reactor(String name) throws IOException {
		this.name = name;
		this.readWriteTask = new ReadWriteTask();
	}

	final void startup() {
		new Thread(readWriteTask, name + "-rw").start();
	}

	/**
	 * 专门启动一个线程进行权限验证。
	 */
	final void postRegister(ISessionContext sessionContext) {
		readWriteTask.registerQueue.offer(sessionContext);
		readWriteTask.selector.wakeup();
	}

	final Queue<ISessionContext> getRegisterQueue() {
		return readWriteTask.registerQueue;
	}

	final long getReactCount() {
		return readWriteTask.reactCount;
	}

	private final class ReadWriteTask implements Runnable {
		private final Selector selector;
		private final ConcurrentLinkedQueue<ISessionContext> registerQueue;
		private long reactCount;

		private ReadWriteTask() throws IOException {
			this.selector = Selector.open();
			this.registerQueue = new ConcurrentLinkedQueue<ISessionContext>();
		}

		@Override
		public void run() {
			final Selector selector = this.selector;
			Set<SelectionKey> selectedKeySet = null;
			for (;;) {
				++reactCount;
				try {
					selector.select(500L);
					register(selector);
					selectedKeySet = selector.selectedKeys();
					for (SelectionKey selectedKey : selectedKeySet) {
						SessionContext$ReadWrite sessionContext = null;
						try {
							Object attachment = selectedKey.attachment();
							if (attachment != null) {
								sessionContext = (SessionContext$ReadWrite) attachment;
								if (selectedKey.isValid() && selectedKey.isReadable()) {
									try {
										sessionContext.read();
									} catch (IOException e) {
										sessionContext.close("program err:" + e.toString());
										continue;
									} catch (Exception e) {
										LOGGER.error("caught err:", e);
										sessionContext.close("program err:" + e.toString());
										continue;
									}
								}
								if (selectedKey.isValid() && selectedKey.isWritable()) {
									sessionContext.write();
								}
							} else {
								selectedKey.cancel();
							}
						} catch (CancelledKeyException e) {
							if (LOGGER.isDebugEnabled()) {
								LOGGER.debug(sessionContext + " socket key canceled");
							}
						} catch (Exception e) {
							LOGGER.error(sessionContext + " " + e);
						} catch (final Throwable e) {
							// Catch exceptions such as OOM and close connection
							// if exists
							// so that the reactor can keep running!
							// @author Uncle-pan
							// @since 2016-03-30
							if (sessionContext != null) {
								sessionContext.close("Bad: " + e);
							}
							LOGGER.error("caught err: ", e);
							continue;
						}
					}
				} catch (Exception e) {
					LOGGER.error(name, e);
				} catch (final Throwable e) {
					// Catch exceptions such as OOM so that the reactor can keep
					// running!
					// @author Uncle-pan
					// @since 2016-03-30
					LOGGER.error("caught err: ", e);
				} finally {
					if (selectedKeySet != null) {
						selectedKeySet.clear();
					}

				}
			}
		}

		private void register(Selector selector) {
			ISessionContext sessionContext = null;
			if (registerQueue.isEmpty()) {
				return;
			}
			while ((sessionContext = registerQueue.poll()) != null) {
				try {
					/**
					 * 注册读数据事件
					 */
					((ReadWrite) sessionContext.getReadWrite()).registerReadEvent(selector);
					sessionContext.init();
					/**
					 * 验证成功之后才能建立SessionContext。
					 */
				} catch (Exception e) {
					sessionContext.close("register err" + e.toString());
				}
			}
		}

	}

}